package org.tao.lightningmq.broker.slave;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.event.model.ConsumeMsgAckEvent;
import org.tao.lightningmq.broker.event.model.CreateTopicEvent;
import org.tao.lightningmq.broker.event.model.PushMsgEvent;
import org.tao.lightningmq.broker.model.ConsumeQueueConsumeAckModel;
import org.tao.lightningmq.common.dto.ConsumeMsgAckReqDTO;
import org.tao.lightningmq.common.dto.CreateTopicReqDTO;
import org.tao.lightningmq.common.dto.MessageDTO;
import org.tao.lightningmq.common.dto.SlaveSyncRespDTO;
import org.tao.lightningmq.common.enums.BrokerEventCode;
import org.tao.lightningmq.common.enums.BrokerResponseCode;
import org.tao.lightningmq.common.event.EventBus;
import org.tao.lightningmq.common.event.model.BaseEvent;
import org.tao.lightningmq.common.remote.SyncFuture;
import org.tao.lightningmq.common.remote.SyncFutureManager;
import org.tao.lightningmq.common.tcp.TcpMsg;

import java.net.InetSocketAddress;
import java.util.Map;

/**
 * @Author lht
 * @date 2025/1/21 - 14:17
 * @description:
 */
public class SlaveSyncServerHandler extends SimpleChannelInboundHandler<TcpMsg> {

    private static final Logger LOGGER = LoggerFactory.getLogger(SlaveSyncServerHandler.class);

    private EventBus eventBus;

    public SlaveSyncServerHandler(EventBus eventBus) {
        this.eventBus = eventBus;
        this.eventBus.init();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TcpMsg tcpMsg) throws Exception {
        int code = tcpMsg.getCode();
        byte[] body = tcpMsg.getBody();
        BaseEvent event = null;
        if (BrokerEventCode.CREATE_TOPIC.getCode() == code) {
            CreateTopicReqDTO createTopicReqDTO = JSON.parseObject(body, CreateTopicReqDTO.class);
            CreateTopicEvent createTopicEvent = new CreateTopicEvent();
            createTopicEvent.setCreateTopicReqDTO(createTopicReqDTO);
            createTopicEvent.setMsgId(createTopicReqDTO.getMsgId());
            event = createTopicEvent;
        } else if (BrokerEventCode.PUSH_MSG.getCode() == code) {
            MessageDTO messageDTO = JSON.parseObject(body, MessageDTO.class);
            PushMsgEvent pushMsgEvent = new PushMsgEvent();
            pushMsgEvent.setMessageDTO(messageDTO);
            pushMsgEvent.setMsgId(messageDTO.getMsgId());
            LOGGER.info("收到消息同步内容:{},message is {}", new String(messageDTO.getBody()), JSON.toJSONString(messageDTO));
            event = pushMsgEvent;
        } else if (BrokerEventCode.CONSUME_SUCCESS_MSG.getCode() == code) {
            ConsumeMsgAckReqDTO consumeMsgAckReqDTO = JSON.parseObject(body, ConsumeMsgAckReqDTO.class);
            String topic = consumeMsgAckReqDTO.getTopic();
            String consumeGroup = consumeMsgAckReqDTO.getConsumeGroup();
            Integer queueId = consumeMsgAckReqDTO.getQueueId();
            Integer ackCount = consumeMsgAckReqDTO.getAckCount();
            Integer startMsgConsumeQueueOffset = consumeMsgAckReqDTO.getStartMsgConsumeQueueOffset();
            ConsumeQueueConsumeAckModel consumeQueueConsumeAckModel =
                    new ConsumeQueueConsumeAckModel(topic, consumeGroup, queueId, ackCount, startMsgConsumeQueueOffset);
            if (CommonCache.getConsumeQueueConsumeHandler().ack(consumeQueueConsumeAckModel)) {
                LOGGER.info("sync information, broker receive offset value ,topic is {},consumeGroup is {},queueId is {},ackCount is {}",
                        topic, consumeGroup, queueId, ackCount);
                // 从节点给主节点回复同步情况
                SlaveSyncRespDTO slaveSyncRespDTO = new SlaveSyncRespDTO();
                slaveSyncRespDTO.setMsgId(consumeMsgAckReqDTO.getMsgId());
                slaveSyncRespDTO.setSyncSuccess(true);
                LOGGER.info("从节点同步消费进度成功");
                channelHandlerContext.writeAndFlush(new TcpMsg(BrokerResponseCode.SLAVE_SYNC_RESP.getCode(),
                        JSON.toJSONBytes(slaveSyncRespDTO)));
            }
        } else {
            Map map = JSON.parseObject(body, Map.class);
            String msgId = (String) map.get("msgId");
            SyncFuture syncFuture = SyncFutureManager.get(msgId);
            if (syncFuture != null) {
                syncFuture.setResp(tcpMsg);
            }
        }
        if (event != null) {
            event.setContext(channelHandlerContext);
            eventBus.publish(event);
        }
    }
}
